package com.atguigu.day03;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;

/**
 * @description :
 * @autor :lyh
 */
public class Example7 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .socketTextStream("localhost", 9999)
                .keyBy(r -> "String")
                .process(new KeyedProcessFunction<String, String, String>() {
                    @Override
                    public void processElement(String in, KeyedProcessFunction<String, String, String>.Context ctx, Collector<String> out) throws Exception {
                        // 获取当前的机器时间（处理时间）
                        long currentTimeStamp = ctx.timerService().currentProcessingTime();
                        long thirtySecondLater = currentTimeStamp + 30 * 1000L;
                        long sixtySecondLater = currentTimeStamp + 60 * 1000L;

                        // 注册30秒钟后的一个定时器
                        // 注册的是输入数据in的key对应的定时器
                        ctx.timerService().registerProcessingTimeTimer(thirtySecondLater);
                        // 注册60秒钟后的一个定时器
                        ctx.timerService().registerProcessingTimeTimer(sixtySecondLater);

                        out.collect("输入数据：" + in + "，key是：" + ctx.getCurrentKey() + "，当前的处理时间是："+
                                "" + new Timestamp(currentTimeStamp) + "，注册的定时器的时间戳是：" +
                                "" + new Timestamp(thirtySecondLater) + "；" +
                                "" + new Timestamp(sixtySecondLater));
                    }

                    // 定时器的具体逻辑
                    // 当机器时间行进到'timerTimeStamp'时，触发
                    @Override
                    public void onTimer(long timerTimestamp, KeyedProcessFunction<String, String, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                        long currentTimeStamp = ctx.timerService().currentProcessingTime();
                        out.collect("定时器的key: "+ctx.getCurrentKey()+"，定时器的时间戳："+
                                "" + new Timestamp(timerTimestamp) + "，定时器真正的执行时间戳是："+
                                "" + new Timestamp(currentTimeStamp));
                    }
                })
                .print();

        env.execute();
    }
}
